package com.fwmagic.flinkcdc.postgres;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @description:flinkcdc for postgres whith flinkSQL
 * @author: fangwei
 * @date: 2022/5/19
 **/
public class FlinkSQLCDC {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String sourceDDL =
                "CREATE TABLE pgsql_source (\n" +
                        " id int,\n" +
                        " name STRING\n" +
                        ") WITH (\n" +
                        " 'connector' = 'postgres-cdc',\n" +
                        " 'hostname' = 'localhost',\n" +
                        " 'port' = '5432',\n" +
                        " 'username' = 'postgres',\n" +
                        " 'password' = '123456',\n" +
                        " 'database-name' = 'postgres',\n" +
                        " 'schema-name' = 'public',\n" +
                        " 'debezium.plugin.name' = 'pgoutput',\n" +
                        " 'table-name' = 't_stu'\n" +
                        ")";


        //创建 Flink-Postgres-CDC
        tableEnv.executeSql(sourceDDL);
        Table table = tableEnv.sqlQuery("select * from pgsql_source");

        DataStream<Tuple2<Boolean, Row>> retractStream = tableEnv.toRetractStream(table, Row.class);

        retractStream.print();


        env.execute("FlinkSQL-CDC postgres binlog");
    }
}
